package io.mft.datasource.mq.container.impl;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Binding.DestinationType;
import org.springframework.amqp.core.ExchangeBuilder;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.ConsumerTagStrategy;

import cn.hutool.core.lang.UUID;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONUtil;
import io.mft.commons.config.mq.MqConsumer;
import io.mft.commons.config.mq.MqExchange;
import io.mft.commons.config.mq.MqQueue;
import io.mft.commons.exception.MFTException;
import io.mft.commons.function.response.kv.KVMap;
import io.mft.commons.util.SequenceGenerator;
import io.mft.datasource.mq.container.MqContainer;
import io.mft.datasource.mq.container.MqContainerType;
import io.mft.datasource.mq.rabbitmq.MqDatasource;
import io.mft.datasource.mq.rabbitmq.admin.AdminConnectionFactory;
import io.mft.datasource.mq.rabbitmq.consumer.MessageHandler;
import io.mft.datasource.mq.rabbitmq.consumer.MqConsumerConnectionFactory;
import io.mft.datasource.mq.rabbitmq.producer.MqProducerConnectionFactory;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class RabbitMqContainer extends MqContainer {
	
   
    Map<String, RabbitAdmin> mapRabbitAdmin=new ConcurrentHashMap<String, RabbitAdmin>();

    Map<String,Map<MqDatasourceType, CachingConnectionFactory>> mapCachingConnectionFactory=new ConcurrentHashMap<String, Map<MqDatasourceType,CachingConnectionFactory>>();
    SequenceGenerator sequenceGenerator=new SequenceGenerator();
    Map<String,Map<String, RabbitTemplate>> mapRabbitTemplate = new ConcurrentHashMap<>();
    @Override
    protected void initedMqDatasource(MqDatasource datasource) {
    	AdminConnectionFactory adminConnectionFactory=new AdminConnectionFactory(datasource);
    	MqProducerConnectionFactory mqProducerConnectionFactory=new MqProducerConnectionFactory(datasource);
    	mqProducerConnectionFactory.afterPropertiesSet();
    	MqConsumerConnectionFactory mqConsumerConnectionFactory=new MqConsumerConnectionFactory(datasource);
    	mqConsumerConnectionFactory.afterPropertiesSet();
    	Map<MqDatasourceType, CachingConnectionFactory> fmap=new ConcurrentHashMap<MqDatasourceType, CachingConnectionFactory>();
    	fmap.put(MqDatasourceType.admin, adminConnectionFactory );
    	fmap.put(MqDatasourceType.producer, mqProducerConnectionFactory);
    	fmap.put(MqDatasourceType.consumer, mqConsumerConnectionFactory);
    	mapCachingConnectionFactory.put(datasource.getId(), fmap);
    	RabbitAdmin rabbitAdmin=new RabbitAdmin(adminConnectionFactory);
    	rabbitAdmin.afterPropertiesSet();
    	mapRabbitAdmin.put(datasource.getId(), rabbitAdmin);
	}
   

    /**
     * 关闭数据源
     * @说明: TODO
     * @方法名称: shutDown
     * @参数 @param datasourceId
     * @返回类型 void    
     * @创建者: 胡常建  
     * @创建时间: 2021年4月23日 下午4:15:38
     * @修改者: 胡常建  
     * @修改时间: 2021年4月23日 下午4:15:38
     */
    @Override
    public void shutDown(String datasourceId) {
		
	}
    

    
   

    protected void addedMqExchange(String pluginId,MqExchange exchange) {
    	RabbitAdmin admin=mapRabbitAdmin.get(exchange.getDatasourceId());
    	org.springframework.amqp.core.Exchange obj=ExchangeBuilder.fanoutExchange(exchange.getId()).durable(true).build();
    	admin.declareExchange(obj);
    	exchange.getBinds().forEach(mqBind->{
    		MqQueue queue=mapMqQueue.get(mqContainerType()).get(pluginId).get(mqBind.getQueueId());
    		if (queue!=null) {
    			String key=exchange.getId()+"."+queue.getName();
    			Binding binding=new Binding(queue.getName(), DestinationType.QUEUE, exchange.getId(), key, MapUtil.newHashMap());
    			admin.declareBinding(binding);
			}
    	});
    	addRabbitTemplate(pluginId,exchange);
	}
    
    private void addRabbitTemplate(String pluginId,MqExchange exchange) {
    	Map<MqDatasourceType, CachingConnectionFactory> map=mapCachingConnectionFactory.get(exchange.getDatasourceId());
    	CachingConnectionFactory mqProducerConnectionFactory=map.get(MqDatasourceType.producer);
       	RabbitTemplate rabbitTemplate=new RabbitTemplate(mqProducerConnectionFactory);
    	rabbitTemplate.setMandatory(true);
    	rabbitTemplate.setReturnCallback(new ReturnCallback() {
			
			@Override
			public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
				
			}
		});
    	rabbitTemplate.afterPropertiesSet();
    	Map<String, RabbitTemplate> m=mapRabbitTemplate.get(pluginId);
    	if (ObjectUtil.isNull(m)) {
			m=new ConcurrentHashMap<>();
			mapRabbitTemplate.put(pluginId, m);
		}
    	m.put(exchange.getId(), rabbitTemplate);
	}
    protected void addedMqQueue(String pluginId,MqQueue queue) {
    	Queue queueObj=QueueBuilder.durable(queue.getName()).build();
    	RabbitAdmin admin=mapRabbitAdmin.get(queue.getDatasourceId());
    	admin.declareQueue(queueObj);
	}
    protected void addedMqConsumer(String pluginId,MqConsumer consumer) {
    	SimpleMessageListenerContainer listenerContainer=new SimpleMessageListenerContainer(mapCachingConnectionFactory.get(consumer.getDatasourceId()).get(MqDatasourceType.consumer));
    	MqQueue queue=getQueue(pluginId,consumer.getQueueId());
    	listenerContainer.setQueueNames(queue.getName());
    	listenerContainer.setConcurrentConsumers(consumer.getConcurrent());
    	listenerContainer.setMaxConcurrentConsumers(consumer.getMaxConcurrent());
    	listenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);
    	listenerContainer.setMessageListener(new MessageHandler(consumer));
    	listenerContainer.setConsumerTagStrategy(new ConsumerTagStrategy() {    // 消费端的标签策略
			@Override
			public String createConsumerTag(String queue) { //为每个动态的消费者定义一个tag
				return queue + "_" + UUID.randomUUID().toString();
			}
		});
    	listenerContainer.afterPropertiesSet();
    	listenerContainer.start();
	}


	@Override
	public void sendMessage(String pluginId,String exchangeId, KVMap message) {
		try {
			RabbitTemplate template = mapRabbitTemplate.get(pluginId).get(exchangeId);
			if (ObjectUtil.isNull(template)) {
				throw new MFTException("交换机[" + exchangeId + "]未初始化RabbitTemplate");
			}
			MessageProperties messageProperties = new MessageProperties();
			messageProperties.setMessageId(UUID.randomUUID().toString());
			//long deliveryTag=sequenceGenerator.nextId();
			//messageProperties.setDeliveryTag(deliveryTag);
			String json = JSONUtil.toJsonStr(message);
			byte[] bytes = json.getBytes("UTF-8");
			Message m = new Message(bytes, messageProperties);
			template.convertAndSend(exchangeId, null, m);
		} catch (Exception e) {
			log.error("", e);
		}
	}


	@Override
	public MqContainerType mqContainerType() {
		return MqContainerType.rabbitmq;
	}
}
